"""基于 DuckDuckGo 的网页搜索工具，用于搜索最新资讯以分析 ETF 风险点."""

from __future__ import annotations

import asyncio
import json
import logging
from typing import Any, Dict, List, Optional

from langchain_community.tools import DuckDuckGoSearchResults

from .timezone import now_shanghai


logger = logging.getLogger(__name__)


def _format_search_text(payload: dict) -> str:
    """将搜索结果格式化为易读的中文文本，便于理解."""
    lines = []
    
    # 标题
    lines.append("【网页搜索结果】")
    lines.append("")
    
    # 查询信息
    lines.append("🔍 查询信息：")
    lines.append(f"  • 查询关键词: {payload.get('query', 'N/A')}")
    lines.append(f"  • 结果数量: {payload.get('result_count', 0)} 条")
    lines.append(f"  • 最近一周内结果: {payload.get('recent_count', 0)} 条")
    if payload.get('avg_relevance_score'):
        lines.append(f"  • 平均相关性分数: {payload.get('avg_relevance_score', 0):.3f}")
    lines.append("")
    
    # 搜索结果
    results = payload.get("results", [])
    if results:
        lines.append("📰 搜索结果：")
        for i, item in enumerate(results[:10], 1):  # 只显示前10个结果
            lines.append(f"\n  {i}. {item.get('title', '无标题')}")
            if item.get('source'):
                lines.append(f"     来源: {item['source']}")
            if item.get('snippet'):
                snippet = item['snippet']
                # 截断过长的摘要
                if len(snippet) > 200:
                    snippet = snippet[:200] + "..."
                lines.append(f"     摘要: {snippet}")
            if item.get('date'):
                lines.append(f"     时间: {item['date']}")
            if item.get('relevance_score'):
                lines.append(f"     相关性: {item['relevance_score']:.3f}")
        if len(results) > 10:
            lines.append(f"\n  ... 还有 {len(results) - 10} 个结果")
    else:
        lines.append("  • 未找到相关结果")
    
    lines.append("")
    
    # 使用提示
    lines.append("💡 提示：")
    lines.append("  • 这些结果可用于分析 ETF 的风险点和市场动态")
    lines.append("  • 建议重点关注最近一周内的最新资讯")
    lines.append("  • 可以结合风险分析工具进行综合评估")
    
    return "\n".join(lines)


class WebSearch:
    """基于 DuckDuckGo 的网页搜索工具，用于搜索最新资讯以分析 ETF 风险点.
    
    这是一个普通的工具类，不依赖 LangChain 框架，可以直接使用。
    提供同步和异步两种搜索方式，支持时间过滤、相关性评分和结果排序。
    """

    def __init__(
        self,
        max_results: Optional[int] = None,
        timeout: int = 30,
        time_range_days: int = 7,
        min_relevance_score: float = 0.1,
        raw_search_results: int = 20,
    ) -> None:
        """初始化搜索工具.

        Args:
            max_results: 最大返回结果数（过滤后），None 表示不限制
            timeout: 搜索超时时间（秒），默认 30 秒
            time_range_days: 时间范围（天），默认 7 天（一周内）
            min_relevance_score: 最小相关性分数（0-1），低于此分数的结果将被过滤，默认 0.1
            raw_search_results: 从搜索引擎获取的原始结果数量，默认 20（增加原始结果以提高过滤后的结果数量）
        """
        # 增加原始搜索结果数量，以便过滤后有更多结果
        # 注意：DuckDuckGoSearchResults 使用 num_results 作为参数别名
        self._search = DuckDuckGoSearchResults(num_results=raw_search_results)
        logger.debug("初始化搜索工具 - raw_search_results=%s, max_results=%s", 
                    raw_search_results, max_results)
        self._max_results = max_results
        self._timeout = timeout
        self._time_range_days = time_range_days
        self._min_relevance_score = min_relevance_score
        self._raw_search_results = raw_search_results

    def _ensure_recent(self, query: str) -> str:
        """在查询中添加时间关键词，确保搜索返回一周内的最新资讯.

        用于 ETF 风险点分析，需要获取最新的市场资讯。

        Args:
            query: 原始查询字符串

        Returns:
            添加了时间关键词的查询字符串
        """
        if not query or not query.strip():
            return query

        query_lower = query.lower()
        
        # 检查是否已包含时间范围关键词
        time_range_keywords = [
            "过去7天", "过去一周", "最近7天", "最近一周", "一周内", "7天内",
            "past 7 days", "past week", "last 7 days", "last week", "within 7 days",
        ]
        has_time_range = any(keyword in query_lower for keyword in time_range_keywords)
        
        # 检查是否已包含一般时间关键词
        time_keywords = ["最新", "newest", "recent", "latest", "最新消息", "最新资讯"]
        has_time_keyword = any(keyword in query_lower for keyword in time_keywords)

        if not has_time_range:
            # 添加明确的一周时间范围关键词
            current_date = now_shanghai().strftime("%Y年%m月%d日")
            # 使用多种时间表达方式提升搜索命中率
            query = f"{query} 过去7天 最近一周 {current_date}"
        elif not has_time_keyword:
            # 如果已有时间范围但没有一般时间关键词，添加"最新"
            query = f"{query} 最新"

        return query

    def _extract_time_info(self, snippet: str) -> Optional[Dict[str, Any]]:
        """从 snippet 中提取时间信息.

        支持格式：
        - 相对时间：7 hours ago, 2 days ago, 3天前, 一周前, 7 hours ago, 2 days ago
        - 绝对时间：2025-11-12, Nov 12, 2025, 11月12日

        Args:
            snippet: 搜索结果摘要

        Returns:
            包含时间信息的字典，格式：{
                "extracted_time": ISO 格式时间字符串或相对时间描述,
                "is_recent": 是否在一周内（布尔值）,
                "time_score": 时间新鲜度分数（0-1）
            }
            如果无法提取时间，返回 None
        """
        import re
        from datetime import datetime, timedelta

        if not snippet:
            return None

        current_time = now_shanghai()
        time_range = timedelta(days=self._time_range_days)
        threshold_time = current_time - time_range

        # 匹配相对时间（英文）
        # 格式：X hours ago, X days ago, X weeks ago, X months ago
        relative_pattern_en = r"(\d+)\s*(hour|hours|day|days|week|weeks|month|months)\s+ago"
        match_en = re.search(relative_pattern_en, snippet, re.IGNORECASE)
        if match_en:
            value = int(match_en.group(1))
            unit = match_en.group(2).lower()
            
            if unit in ["hour", "hours"]:
                delta = timedelta(hours=value)
            elif unit in ["day", "days"]:
                delta = timedelta(days=value)
            elif unit in ["week", "weeks"]:
                delta = timedelta(weeks=value)
            elif unit in ["month", "months"]:
                delta = timedelta(days=value * 30)  # 近似值
            else:
                return None

            extracted_time = current_time - delta
            is_recent = extracted_time >= threshold_time
            
            # 计算时间新鲜度分数（越新分数越高）
            if is_recent:
                days_diff = (current_time - extracted_time).days
                time_score = max(0, 1.0 - (days_diff / self._time_range_days))
            else:
                time_score = 0.0

            return {
                "extracted_time": extracted_time.isoformat(),
                "is_recent": is_recent,
                "time_score": time_score,
                "time_description": match_en.group(0),
            }

        # 匹配相对时间（中文）
        # 格式：X小时前, X天前, X周前, X个月前
        relative_pattern_cn = r"(\d+)\s*(小时|天|周|个月|月)\s*前"
        match_cn = re.search(relative_pattern_cn, snippet)
        if match_cn:
            value = int(match_cn.group(1))
            unit = match_cn.group(2)

            if unit == "小时":
                delta = timedelta(hours=value)
            elif unit == "天":
                delta = timedelta(days=value)
            elif unit == "周":
                delta = timedelta(weeks=value)
            elif unit in ["个月", "月"]:
                delta = timedelta(days=value * 30)  # 近似值
            else:
                return None

            extracted_time = current_time - delta
            is_recent = extracted_time >= threshold_time
            
            if is_recent:
                days_diff = (current_time - extracted_time).days
                time_score = max(0, 1.0 - (days_diff / self._time_range_days))
            else:
                time_score = 0.0

            return {
                "extracted_time": extracted_time.isoformat(),
                "is_recent": is_recent,
                "time_score": time_score,
                "time_description": match_cn.group(0),
            }

        # 匹配绝对日期（ISO 格式：YYYY-MM-DD）
        iso_pattern = r"(\d{4})-(\d{1,2})-(\d{1,2})"
        match_iso = re.search(iso_pattern, snippet)
        if match_iso:
            try:
                year, month, day = int(match_iso.group(1)), int(match_iso.group(2)), int(match_iso.group(3))
                extracted_time = datetime(year, month, day, tzinfo=current_time.tzinfo)
                is_recent = extracted_time >= threshold_time
                
                if is_recent:
                    days_diff = (current_time - extracted_time).days
                    time_score = max(0, 1.0 - (days_diff / self._time_range_days))
                else:
                    time_score = 0.0

                return {
                    "extracted_time": extracted_time.isoformat(),
                    "is_recent": is_recent,
                    "time_score": time_score,
                    "time_description": match_iso.group(0),
                }
            except ValueError:
                pass

        # 如果没有找到时间信息，返回 None
        return None

    def _calculate_relevance_score(self, item: Dict[str, Any], query: str) -> float:
        """计算搜索结果的相关性分数.

        考虑因素：
        - 标题中的关键词匹配（权重更高）
        - 摘要中的关键词匹配
        - 关键词出现频率
        - 关键词位置（标题权重更高）

        Args:
            item: 搜索结果项
            query: 原始查询字符串

        Returns:
            相关性分数（0-1），分数越高越相关
        """
        if not query or not query.strip():
            return 0.5  # 如果没有查询，返回中等分数

        # 提取查询关键词（去除时间相关关键词）
        query_lower = query.lower()
        time_keywords = [
            "过去7天", "过去一周", "最近7天", "最近一周", "一周内", "7天内",
            "最新", "newest", "recent", "latest", "最新消息", "最新资讯",
            "past 7 days", "past week", "last 7 days", "last week",
        ]
        
        # 移除时间关键词，只保留核心查询词
        core_query = query_lower
        for keyword in time_keywords:
            core_query = core_query.replace(keyword, " ")
        
        # 移除日期格式（如 "2025年11月12日"）
        import re
        # 移除中文日期格式：2025年11月12日、2025年11月、11月12日等
        core_query = re.sub(r"\d{4}年\d{1,2}月\d{1,2}日", " ", core_query)
        core_query = re.sub(r"\d{4}年\d{1,2}月", " ", core_query)
        core_query = re.sub(r"\d{1,2}月\d{1,2}日", " ", core_query)
        # 移除单独的"年"、"月"、"日"
        core_query = re.sub(r"[年月日]", " ", core_query)
        
        # 提取关键词（去除停用词和标点）
        keywords = [w for w in re.split(r"\s+", core_query.strip()) if len(w) > 1]
        
        if not keywords:
            # 如果没有关键词，返回一个基础分数而不是0，避免全部被过滤
            logger.debug("无法提取关键词，查询: %s, 处理后: %s", query, core_query)
            return 0.2

        title = (item.get("title") or "").lower()
        snippet = (item.get("snippet") or "").lower()
        
        # 记录关键词提取结果（仅在调试模式下）
        if logger.isEnabledFor(logging.DEBUG):
            logger.debug("关键词提取 - 原始查询: %s, 核心查询: %s, 关键词: %s", 
                        query[:50], core_query[:50], keywords)
        
        # 计算标题中的匹配分数（权重 0.6）
        title_score = 0.0
        title_matches = sum(1 for keyword in keywords if keyword in title)
        if title_matches > 0:
            title_score = min(1.0, title_matches / len(keywords)) * 0.6

        # 计算摘要中的匹配分数（权重 0.4）
        snippet_score = 0.0
        snippet_matches = sum(1 for keyword in keywords if keyword in snippet)
        if snippet_matches > 0:
            snippet_score = min(1.0, snippet_matches / len(keywords)) * 0.4

        # 综合分数
        relevance_score = title_score + snippet_score
        
        # 如果标题和摘要都包含关键词，给予额外加分
        if title_matches > 0 and snippet_matches > 0:
            relevance_score = min(1.0, relevance_score + 0.1)
        
        # 如果完全没有匹配，给予一个基础分数（0.1），避免因为完全没匹配而被过滤
        # 这样可以保留一些可能相关但关键词不完全匹配的结果
        if relevance_score == 0.0:
            relevance_score = 0.1

        return relevance_score

    def _sort_by_relevance_and_time(
        self, results: List[Dict[str, Any]], query: str
    ) -> List[Dict[str, Any]]:
        """按相关性和时间排序结果.

        排序规则：
        1. 优先显示一周内的结果
        2. 在一周内的结果中，按相关性分数排序
        3. 相关性相同时，按时间新鲜度排序

        Args:
            results: 搜索结果列表
            query: 原始查询字符串

        Returns:
            排序后的结果列表
        """
        # 为每个结果计算相关性分数
        for item in results:
            if "relevance_score" not in item:
                item["relevance_score"] = self._calculate_relevance_score(item, query)
            
            # 如果没有时间分数，设置为 0.5（中等）
            if "time_score" not in item:
                item["time_score"] = 0.5

        # 排序：优先一周内的结果，然后按综合分数排序
        def sort_key(item: Dict[str, Any]) -> tuple:
            is_recent = item.get("is_recent", False)
            relevance_score = item.get("relevance_score", 0.0)
            time_score = item.get("time_score", 0.5)
            
            # 综合分数 = 相关性 * 0.7 + 时间新鲜度 * 0.3
            combined_score = relevance_score * 0.7 + time_score * 0.3
            
            # 返回排序键：(是否一周内, 综合分数)
            # 一周内的结果排在前面，然后按综合分数降序
            return (not is_recent, -combined_score)

        return sorted(results, key=sort_key)

    def _filter_by_time_range(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """过滤掉超过时间范围的结果，并为结果添加时间相关字段.

        注意：如果所有结果都被时间过滤掉，则放宽限制，保留所有结果（但标记为不在一周内）。
        这样可以避免因为时间过滤过于严格而导致完全没有结果。

        Args:
            results: 搜索结果列表

        Returns:
            过滤后的结果列表，每个结果包含时间相关字段
        """
        filtered_results = []
        filtered_out_results = []
        
        for item in results:
            snippet = item.get("snippet", "")
            time_info = self._extract_time_info(snippet)
            
            if time_info:
                # 添加时间信息
                item["extracted_time"] = time_info.get("extracted_time")
                item["is_recent"] = time_info.get("is_recent", False)
                item["time_score"] = time_info.get("time_score", 0.0)
                item["time_description"] = time_info.get("time_description")
                
                # 如果在一周内，保留结果
                if time_info.get("is_recent", False):
                    filtered_results.append(item)
                else:
                    # 暂时保存被过滤的结果，如果最终没有结果，会重新加入
                    filtered_out_results.append(item)
            else:
                # 如果无法提取时间信息，默认保留（可能是最新内容）
                # 但给予较低的时间分数
                item["extracted_time"] = None
                item["is_recent"] = True  # 默认假设是最近的
                item["time_score"] = 0.3  # 较低的时间分数
                item["time_description"] = None
                filtered_results.append(item)

        # 如果时间过滤后完全没有结果，放宽限制，保留所有结果
        # 这样可以避免因为时间过滤过于严格而导致完全没有结果
        if not filtered_results and filtered_out_results:
            logger.warning("时间过滤后没有结果，放宽限制保留所有结果（共 %s 个）", len(filtered_out_results))
            for item in filtered_out_results:
                # 保留结果但标记为不在一周内，时间分数设为 0
                item["is_recent"] = False
                item["time_score"] = 0.0
                filtered_results.append(item)

        return filtered_results

    def _parse_formatted_string(self, text: str) -> List[Dict[str, Any]]:
        """解析 DuckDuckGo 返回的格式化字符串.

        DuckDuckGo 有时返回格式如下的字符串：
        snippet: ..., title: ..., link: ..., snippet: ..., title: ..., link: ...

        Args:
            text: 格式化字符串

        Returns:
            解析后的结果列表
        """
        import re

        results = []
        # 使用正则表达式匹配 title、snippet、link 字段
        # 匹配模式：field: value（字段名后跟冒号和空格，然后匹配到下一个字段名或字符串结尾）
        # 使用非贪婪匹配，直到遇到下一个字段名
        pattern = r"(title|snippet|link):\s*([^,]+?)(?=\s*,\s*(?:title|snippet|link):|$)"
        matches = re.findall(pattern, text, re.IGNORECASE | re.DOTALL)

        current_item: Dict[str, Any] = {}
        for field, value in matches:
            field_lower = field.lower().strip()
            value = value.strip().rstrip(",")  # 移除末尾可能的逗号
            if value:
                current_item[field_lower] = value

            # 当遇到 link 字段时，表示一个完整结果结束
            if field_lower == "link" and current_item:
                if current_item.get("title") or current_item.get("snippet") or current_item.get("link"):
                    results.append(current_item.copy())
                    current_item = {}

        # 处理最后一个未完成的结果（如果没有 link 字段）
        if current_item and (current_item.get("title") or current_item.get("snippet") or current_item.get("link")):
            results.append(current_item)

        return results

    def _parse_results(self, results: List[Dict[str, Any]], query: str) -> List[Dict[str, Any]]:
        """解析搜索结果，过滤空结果，应用时间过滤和相关性过滤.

        Args:
            results: 原始搜索结果列表
            query: 原始查询字符串（用于相关性评分）

        Returns:
            解析后的结果列表，包含时间信息和相关性分数
        """
        logger.info("开始解析结果，原始结果数量: %s", len(results))
        
        # 基本解析和过滤空结果
        parsed = [
            {
                "title": item.get("title"),
                "snippet": item.get("snippet"),
                "link": item.get("link"),
            }
            for item in results
            if item.get("title") or item.get("snippet") or item.get("link")
        ]
        logger.info("基本解析后结果数量: %s (过滤空结果后)", len(parsed))

        # 应用时间过滤
        before_time_filter = len(parsed)
        parsed = self._filter_by_time_range(parsed)
        after_time_filter = len(parsed)
        if before_time_filter != after_time_filter:
            logger.info("时间过滤后结果数量: %s (过滤前: %s, 被过滤: %s)", 
                       after_time_filter, before_time_filter, 
                       before_time_filter - after_time_filter)
        else:
            logger.debug("时间过滤后结果数量: %s (无变化)", after_time_filter)

        # 计算相关性分数并过滤低相关性结果
        relevance_scores = []
        for item in parsed:
            score = self._calculate_relevance_score(item, query)
            item["relevance_score"] = score
            relevance_scores.append(score)
            # 如果分数为0，记录详细信息用于调试
            if score == 0.0 and logger.isEnabledFor(logging.DEBUG):
                logger.debug("相关性分数为0 - 查询: %s, 标题: %s, 摘要: %s", 
                           query[:50], item.get("title", "")[:50], item.get("snippet", "")[:100])
        
        if relevance_scores:
            logger.info("相关性分数统计 - 最小: %.3f, 最大: %.3f, 平均: %.3f, 阈值: %.3f", 
                        min(relevance_scores), max(relevance_scores), 
                        sum(relevance_scores) / len(relevance_scores), 
                        self._min_relevance_score)

        # 过滤低相关性结果
        before_relevance_filter = len(parsed)
        parsed = [
            item
            for item in parsed
            if item.get("relevance_score", 0.0) >= self._min_relevance_score
        ]
        after_relevance_filter = len(parsed)
        if before_relevance_filter != after_relevance_filter:
            logger.info("相关性过滤后结果数量: %s (过滤前: %s, 被过滤: %s)", 
                       after_relevance_filter, before_relevance_filter, 
                       before_relevance_filter - after_relevance_filter)
        else:
            logger.debug("相关性过滤后结果数量: %s (无变化)", after_relevance_filter)

        # 按相关性和时间排序
        parsed = self._sort_by_relevance_and_time(parsed, query)

        # 应用结果数量限制
        if self._max_results is not None and len(parsed) > self._max_results:
            logger.info("应用结果数量限制: %s -> %s", len(parsed), self._max_results)
            parsed = parsed[: self._max_results]

        logger.info("最终结果数量: %s", len(parsed))
        return parsed

    def _validate_query(self, query: str) -> None:
        """验证查询字符串.

        Args:
            query: 查询字符串

        Raises:
            ValueError: 如果查询字符串无效
        """
        if not query or not query.strip():
            raise ValueError("查询字符串不能为空")
        if len(query.strip()) > 500:
            raise ValueError("查询字符串长度不能超过 500 个字符")

    def search(self, query: str) -> Dict[str, Any]:
        """执行搜索并返回字典格式的结果.

        Args:
            query: 搜索查询字符串

        Returns:
            包含搜索结果的字典，格式：{
                "query": 实际查询字符串,
                "timestamp": ISO 格式时间戳,
                "results": 结果列表,
                "result_count": 结果数量,
                "recent_count": 一周内结果数量,
                "time_range_days": 时间范围（天）,
                "avg_relevance_score": 平均相关性分数
            }

        Raises:
            ValueError: 如果查询字符串无效
            RuntimeError: 如果搜索失败
        """
        # 输入验证
        self._validate_query(query)

        # 添加时间关键词以确保获取最新资讯
        actual_query = self._ensure_recent(query)
        current_time = now_shanghai()

        logger.info("web_search_tool 执行查询：%s", actual_query)

        try:
            # 执行搜索
            results = self._search.run(actual_query)

            # 解析结果
            if isinstance(results, str):
                # 检查是否为空字符串
                if not results.strip():
                    logger.warning("搜索结果为空字符串")
                    raw_items = []
                else:
                    try:
                        raw_items = json.loads(results)
                        logger.debug("成功解析 JSON 结果，类型: %s, 长度: %s", type(raw_items), len(raw_items) if isinstance(raw_items, list) else "N/A")
                    except json.JSONDecodeError:
                        # DuckDuckGo 可能返回格式化字符串而非 JSON
                        logger.debug("尝试解析格式化字符串结果，原始结果长度: %s", len(results))
                        parsed_items = self._parse_formatted_string(results)
                        if parsed_items:
                            raw_items = parsed_items
                            logger.debug("格式化字符串解析成功，解析出 %s 个结果", len(parsed_items))
                        else:
                            # 如果解析失败，将整个字符串作为单个结果
                            logger.debug("格式化字符串解析失败，使用原始结果")
                            raw_items = [{"title": None, "snippet": results, "link": None}]
            else:
                raw_items = results
                logger.debug("搜索结果不是字符串，类型: %s", type(raw_items))

            # 处理结果
            if not isinstance(raw_items, list):
                logger.warning("搜索结果格式异常，转换为列表，原始类型: %s", type(raw_items))
                raw_items = [raw_items] if raw_items else []

            logger.info("原始搜索结果数量: %s", len(raw_items))
            if raw_items:
                logger.debug("原始结果示例（前3个）: %s", raw_items[:3])

            # 解析、过滤和排序结果（传入原始查询用于相关性评分）
            parsed = self._parse_results(raw_items, query)
            
            logger.info("过滤后结果数量: %s (原始: %s)", len(parsed), len(raw_items))

            # 统计信息
            recent_count = sum(1 for item in parsed if item.get("is_recent", False))
            avg_relevance = (
                sum(item.get("relevance_score", 0.0) for item in parsed) / len(parsed)
                if parsed
                else 0.0
            )

            payload = {
                "query": actual_query,
                "timestamp": current_time.isoformat(),
                "results": parsed,
                "result_count": len(parsed),
                "recent_count": recent_count,
                "time_range_days": self._time_range_days,
                "avg_relevance_score": round(avg_relevance, 3),
            }

            logger.info(
                "web_search_tool 查询完成：query=%s, result_count=%s",
                actual_query,
                len(parsed),
            )

            return payload

        except Exception as e:
            logger.error("web_search_tool 搜索失败：query=%s, error=%s", actual_query, e, exc_info=True)
            raise RuntimeError(f"搜索失败：{str(e)}") from e

    def search_json(self, query: str) -> str:
        """执行搜索并返回 JSON 格式的字符串结果.

        Args:
            query: 搜索查询字符串

        Returns:
            JSON 格式的搜索结果字符串

        Raises:
            ValueError: 如果查询字符串无效
            RuntimeError: 如果搜索失败
        """
        result = self.search(query)
        return json.dumps(result, ensure_ascii=False)

    def search_text(self, query: str) -> str:
        """执行搜索并返回格式化的文本结果.

        Args:
            query: 搜索查询字符串

        Returns:
            格式化的中文文本结果

        Raises:
            ValueError: 如果查询字符串无效
            RuntimeError: 如果搜索失败
        """
        result = self.search(query)
        return _format_search_text(result)

    async def search_async(self, query: str) -> Dict[str, Any]:
        """异步执行搜索并返回字典格式的结果.

        Args:
            query: 搜索查询字符串

        Returns:
            包含搜索结果的字典

        Raises:
            ValueError: 如果查询字符串无效
            RuntimeError: 如果搜索失败
        """
        # 使用线程池执行同步搜索，避免阻塞事件循环
        return await asyncio.to_thread(self.search, query)

    # 为了向后兼容，保留 run 方法（返回 JSON 字符串）
    def run(self, query: str) -> str:
        """执行搜索并返回 JSON 格式的字符串结果（向后兼容方法）.

        此方法是为了兼容旧代码而保留的，建议使用 search() 或 search_json() 方法。

        Args:
            query: 搜索查询字符串

        Returns:
            JSON 格式的搜索结果字符串
        """
        return self.search_json(query)

    async def _arun(self, query: str) -> str:
        """异步执行搜索并返回 JSON 格式的字符串结果（向后兼容方法）.

        此方法是为了兼容旧代码而保留的，建议使用 search_async() 方法。

        Args:
            query: 搜索查询字符串

        Returns:
            JSON 格式的搜索结果字符串
        """
        result = await self.search_async(query)
        return json.dumps(result, ensure_ascii=False)

